package org.atguigu.gmall.realtime.join;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * @Author zenghaifeng
 * @Date 2023/4/25 0:10
 * @Package: org.atguigu.gmall.realtime.join
 * @Class: InnerJoinToKafka
 * @Description: TODO
 * @Version 1.0
 */
public class InnerJoinToKafka extends BaseSQLApp {
    public static void main(String[] args) {
        new InnerJoinToKafka().init(50000,2,"InnerJoinToKafka");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        tbe.executeSql("create table a(" +
                " id string, " +
                " name string " +
                ")" + SQLUtil.getKafkaSourceDDL("a", "InnerJoinToKafka_1", "csv"));

        tbe.executeSql("create table b(" +
                " id string, " +
                " age int " +
                ")" + SQLUtil.getKafkaSourceDDL("b", "InnerJoinToKafka_2", "csv"));

        Table result = tbe.sqlQuery("select " +
                " a.id id," +
                " name, " +
                " age " +
                "from a " +
                "join b on a.id=b.id");
        tbe.executeSql("create table ab(" +
                "  id string, " +
                "  name string, " +
                "  age int " +
                ")" + SQLUtil.getKafkaSinkDDL("ab", "json"));

        result.executeInsert("ab");

        //tbe.executeSql("insert into ab select * from " + result);
    }
}
